package com.lagou.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.function.BiConsumer;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: 位移量管理
 * @date 2022/6/11 12:42
 */
public class MyOffsetManager {

    public static void main(String[] args) {
        Map<String,Object> configs = new HashMap<>(4);
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.106.130:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
        // 很重要
        configs.put(ConsumerConfig.GROUP_ID_CONFIG,"mygrp1");
        KafkaConsumer<String,String> consumer = new KafkaConsumer<>(configs);
        // 该方法是懒执行的,只有你的消费者进行poll操作的时候才会把消息均匀的分配给这几个分区
        //consumer.subscribe(Collections.singleton("tp_demo_01"));


        // 当前消费者消费多个主题,那么如何手动的给消费者分配分区
        // 1.需要知道有哪些主题可以访问和消费
        /*Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();
        stringListMap.forEach(new BiConsumer<String, List<PartitionInfo>>() {
            @Override
            public void accept(String s, List<PartitionInfo> partitionInfos) {
                System.out.println("主题名称是"+s);
                partitionInfos.forEach(e->{
                    System.out.println(e);
                });
            }
        });*/

        // 给当前消费者分配指定的主题分区
        consumer.assign(Arrays.asList(
                new TopicPartition("tp_demo_01",0),
                new TopicPartition("tp_demo_01",1),
                new TopicPartition("tp_demo_01",2)
        ));

        // long tp_demo_01 = consumer.position(new TopicPartition("tp_demo_01", 0));
        // System.out.println("当前消费者在0号分区上的偏移量"+tp_demo_01);
        /*consumer.seekToBeginning(Arrays.asList(
                new TopicPartition("tp_demo_01",0),
                new TopicPartition("tp_demo_01",1)
        ));*/

        /*consumer.seekToEnd(Arrays.asList(

                new TopicPartition("tp_demo_01",1)
        ));*/
        System.out.println(consumer.position(new TopicPartition("tp_demo_01", 0)));
        System.out.println(consumer.position(new TopicPartition("tp_demo_01", 1)));
        System.out.println(consumer.position(new TopicPartition("tp_demo_01", 2)));

        consumer.close();

    }
}